package com.jianying.func;

import com.jianying.beans.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * ClassName: MySinkFunction
 * Package: com.atguigu.func
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/5 10:53
 * @Version: 1.0
 */
public class MySinkFunction extends RichSinkFunction<WaterSensor> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 建立连接
        connection = DriverManager.getConnection(
                "jdbc:mysql://localhost:13317/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8",
                "root",
                "123456"
        );
    }

    @Override
    public void invoke(WaterSensor ws) throws Exception {
        // 创建数据库操作对象
        PreparedStatement ps = connection.prepareStatement("insert into ws values(?,?,?)");
        ps.setString(1, ws.getId());
        ps.setLong(2, ws.getTs());
        ps.setInt(3, ws.getVc());

        // 执行sql
        ps.execute();
        // 处理结果集
        // 释放资源
        if(ps != null) {
            ps.close();
        }
    }

    @Override
    public void close() throws Exception {
        if(connection != null) {
            connection.close();
        }
    }
}
